package com.iot08.api.kafka;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Map;


/**
 * @version 1.0
 * @Description: kafka生产者
 * @time: 2020/9/21 14:43
 * @author: Rock
 */
@Slf4j
@Component
public class KafkaProducer {

    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;

    //推送给的push 模块的主题
    @Value("${kafka.producer.topic-name}")
    private String topicName;
    //推送给的设备消息的 主题
    @Value("${kafka.producer.topic-name-watch}")
    private String topicNameWatch;

    //推送给的设备消息的 主题
    @Value("${kafka.producer.topic-name-watch-format}")
    private String topicNameWatchFormat;

    //推送给轨迹纠偏 主题
    @Value("${kafka.producer.topic-name-position-correction}")
    private String topicNamePositionCorrection;

    // 要上传数据到 联通DMP 平台的主题
    @Value("${kafka.producer.topic-name-dmpDevice}")
    private String topicNameDMPDevice;

    // 新增：推送给MQTT服务的专用topic
    @Value("${kafka.producer.topic-name-mqtt:jianyou_tongxin_MQTT_Command}")
    private String topicNameMqtt;
    /**
     * 推送kafka消息
     *
     * @param message
     * @time: 2021/2/22 9:17
     * @author: Rock
     * @version 1.0
     */
    public void send(String message) {
        log.info(String.format("KafkaProducer send. topic: %s, message: %s", topicName, message));
        kafkaTemplate.send(topicName, message);
        kafkaTemplate.flush();
    }

    /**
     * 推送kafka消息 到推送主题
     *
     * @param key
     * @param value
     * @time: 2021/2/22 9:16
     * @author: Rock
     * @version 1.0
     */
    public void send(String key, Object value) {
        log.info(String.format("KafkaProducer send. topic: %s, key: %s, value: %s", topicName, key, value));
        kafkaTemplate.send(topicName, key, value);
        kafkaTemplate.flush();
    }

    /**
     * @param key
     * @param value
     * @time: 2021/1/29 10:10
     * @author: Rock
     * @version 1.0
     */
    public void sendWatch(String key, Object value) {
        log.info(String.format("KafkaProducer send. topic: %s, key: %s, value: %s", topicNameWatch, key, value));
        kafkaTemplate.send(topicNameWatch, key, value);
        kafkaTemplate.flush();
    }

    /**
     * 下发消息 根据imei找对应的标记的kafka topic 下发对应的主题
     *
     * @param key
     * @param value
     * @param kafkaTopicIdent
     * @time: 2021/05/17 17:10
     * @author: WeiChao
     * @version 1.0
     */
    public void sendWatch(String key, Object value, String kafkaTopicIdent) {
        String topic = topicNameWatchFormat + kafkaTopicIdent;
        log.info(String.format("KafkaProducer send. 下发消息 根据imei找对应的标记的kafka topic 下发对应的主题 topic: %s, key: %s, value: %s", topic, key, value));
        kafkaTemplate.send(topic, key, value);
        kafkaTemplate.flush();
    }

    /**
     * @param key
     * @param value
     * @time: 2021/5/19 18:01
     * @author: Rock
     * @version 1.0
     */
    public void sendPositionCorrection(String key, Object value) {
        log.info(String.format("KafkaProducer send. topic: %s, key: %s, value: %s", topicNamePositionCorrection, key, value));
        kafkaTemplate.send(topicNamePositionCorrection, key, value);
        kafkaTemplate.flush();
    }


    public void sendDMPDeviceWithMsg(String key, Object value) {
        log.info(String.format("KafkaProducer send. topic: %s, key: %s, value: %s", topicNameDMPDevice, key, value));
        kafkaTemplate.send(topicNameDMPDevice, key, value);
        kafkaTemplate.flush();
    }


    /**
     * 新增：发送消息到MQTT服务
     * @param imei 设备IMEI
     * @param command 命令类型
     * @param data 数据内容
     */
    public void sendToMqttService(String imei, String command, Object data) {
        try {
            JSONObject message = new JSONObject();
            message.put("command", command);
            message.put("imei", imei);
            message.put("data", data);
            message.put("timestamp", System.currentTimeMillis());
            message.put("source", "api");

            log.info("Sending to MQTT service - topic: {}, imei: {}, command: {}, data: {}",
                    topicNameMqtt, imei, command, data);

            kafkaTemplate.send(topicNameMqtt, imei, message.toJSONString());
            kafkaTemplate.flush();
        } catch (Exception e) {
            log.error("Error sending message to MQTT service for device: {}", imei, e);
        }
    }

    /**
     * 批量发送消息到MQTT服务
     */
    public void batchSendToMqttService(Map<String, Object> messages) {
        try {
            for (Map.Entry<String, Object> entry : messages.entrySet()) {
                String imei = entry.getKey();
                Object data = entry.getValue();
                sendToMqttService(imei, "batch_command", data);
            }
            log.info("Batch sent {} messages to MQTT service", messages.size());
        } catch (Exception e) {
            log.error("Error in batch send to MQTT service", e);
        }
    }

}
